package com.swak.reactivex.transport;

import static java.util.concurrent.TimeUnit.MILLISECONDS;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.Constants;
import com.swak.exception.RemotingException;
import com.swak.exception.ServerException;
import com.swak.reactivex.transport.codec.MultipleCodec;
import com.swak.reactivex.transport.resources.LoopResources;
import com.swak.registry.URL;
import com.swak.utils.StringUtils;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.timeout.IdleStateHandler;

/**
 * use Exchangers.bind();
 * 
 * @author 618lf
 */
public class NettyServer extends Endpoint implements Server {

	protected Logger logger = LoggerFactory.getLogger(NettyServer.class);

	private Channel channel;
	private ServerBootstrap bootstrap;
	private InetSocketAddress localAddress;
	private InetSocketAddress bindAddress;
	private final Map<String, com.swak.reactivex.transport.Channel> channels;
	private final LoopResources resources;

	public NettyServer(URL url, LoopResources resources, ChannelHandler handler) throws RemotingException {
		super(url, handler);
		this.localAddress = getUrl().toInetSocketAddress();
		String bindIp = getUrl().getParameter(Constants.Bind_Host);
		if (StringUtils.isBlank(bindIp)) {
			bindIp = getUrl().getHost();
		}
		this.bindAddress = new InetSocketAddress(bindIp, getUrl().getPort());
		this.channels = new ConcurrentHashMap<>();
		this.resources = resources;

		try {
			this.start();
			if (logger.isInfoEnabled()) {
				logger.info(
						"Start " + getClass().getSimpleName() + " bind " + bindAddress + ", export " + localAddress);
			}
		} catch (Throwable t) {
			throw new RemotingException("Failed to bind " + getClass().getSimpleName() + " on " + localAddress
					+ ", cause: " + t.getMessage(), t);
		}
	}

	@Override
	public Collection<com.swak.reactivex.transport.Channel> getChannels() {
		return channels.values();
	}

	@Override
	public void start() throws ServerException {
		if (this.started) {
			return;
		}
		this.started = true;
		bootstrap = new ServerBootstrap();
		bootstrap.option(ChannelOption.SO_REUSEADDR, Boolean.TRUE) //
				.childOption(ChannelOption.TCP_NODELAY, Boolean.TRUE) //
				.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);//

		// 配置： ServerBootstrap
		this.customBootstrap(bootstrap);

		bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
			@Override
			public void initChannel(SocketChannel ch) throws Exception {
				ChannelPipeline pipeline = ch.pipeline();
				NettyServer.this.customChannel(ch);
				pipeline.addLast("server-idle", new IdleStateHandler(0, 0, getIdleTimeout(), MILLISECONDS));
				pipeline.addLast("handler", new NettyServerHandler(getUrl(), NettyServer.this, channels));
			}
		});
		ChannelFuture channelFuture = bootstrap.bind(this.bindAddress);
		channelFuture.syncUninterruptibly();
		this.channel = channelFuture.channel();
	}

	@Override
	public void close() {
		if (logger.isInfoEnabled()) {
			logger.info("Close " + getClass().getSimpleName() + " bind " + bindAddress);
		}
		try {
			super.close();
		} catch (Throwable e) {
			logger.warn(e.getMessage(), e);
		}
		try {
			doClose();
		} catch (Throwable e) {
			logger.warn(e.getMessage(), e);
		}
	}

	protected void doClose() throws Throwable {
		try {
			if (channel != null) {
				// unbind.
				channel.close();
			}
		} catch (Throwable e) {
			logger.warn(e.getMessage(), e);
		}
		try {
			if (channels != null && channels.size() > 0) {
				for (com.swak.reactivex.transport.Channel channel : channels.values()) {
					try {
						channel.close();
					} catch (Throwable e) {
						logger.warn(e.getMessage(), e);
					}
				}
			}
		} catch (Throwable e) {
			logger.warn(e.getMessage(), e);
		}
		try {
			if (channels != null) {
				channels.clear();
			}
		} catch (Throwable e) {
			logger.warn(e.getMessage(), e);
		}
	}

	@Override
	public void send(Object message, boolean sent) throws RemotingException {
		Collection<com.swak.reactivex.transport.Channel> channels = this.channels.values();
		for (com.swak.reactivex.transport.Channel channel : channels) {
			if (channel.isConnected()) {
				channel.send(message, sent);
			}
		}
	}

	@Override
	public SocketAddress localAddress() {
		return localAddress;
	}

	@Override
	public SocketAddress remoteAddress() {
		return bindAddress;
	}

	@Override
	public void connected(com.swak.reactivex.transport.Channel ch) throws RemotingException {
		// If the server has entered the shutdown process, reject any new connection
		if (this.isClosing() || this.isClosed()) {
			logger.warn("Close new channel " + ch
					+ ", cause: server is closing or has been closed. For example, receive a new connect request while in shutdown process.");
			ch.close();
			return;
		}
		super.connected(ch);
	}

	/**
	 * 配置编解码协议
	 * 
	 * @param ch
	 */
	protected void customChannel(SocketChannel ch) {
		ChannelPipeline pipeline = ch.pipeline();
		Codec codec = new MultipleCodec(getUrl());
		NettyCodec nettyCodec = new NettyCodec(codec, getUrl(), this);
		pipeline.addLast(NettyCodec.Encoder_Name, nettyCodec.getEncoder());
		pipeline.addLast(NettyCodec.Decoder_Name, nettyCodec.getDecoder());
	}

	/**
	 * 配置事件组，配置各种参数
	 * 
	 * @param bootstrap
	 */
	protected void customBootstrap(ServerBootstrap bootstrap) {
		bootstrap.group(resources.onServerSelect(), resources.onServer()).channel(resources.onServerChannel());
	}
}
